package com.atguigu.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.annotation.Nosink;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.bean.VisitorStats;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;


import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Properties;

public class FlinkSinkUtil {
    public static SinkFunction<String> getKafkaSink(String topic) {

        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms",15*60*1000);
        return new FlinkKafkaProducer<String>(
                topic,
                new KafkaSerializationSchema<String>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String value, @Nullable Long aLong) {
                        return new ProducerRecord<>(topic,value.getBytes(StandardCharsets.UTF_8));
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );

    }

    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        return new PhoenixSink();
    }

    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms",15*60*1000);
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
                "default",
                new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element,
                                                                    @Nullable Long aLong) {
                        String topic = element.f1.getSink_table();
                        byte[] data = element.f0.toJSONString().getBytes(StandardCharsets.UTF_8);
                        // topic 和 数据 写入kafka
                        return new ProducerRecord<>(topic,data);
                    }
                },props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

    public static void main(String[] args) {
        getClickHouseSink("aa","dd", VisitorStats.class);
    }
    //通过反射进行,对不同类型的数据进行写入到clickhouse
    public static <T>SinkFunction<T> getClickHouseSink(String db, String table,Class<T> tClass) {
        String driver= "ru.yandex.clickhouse.ClickHouseDriver";
        String url = "jdbc:clickhouse://hadoop162:8123/"+db;
        StringBuilder sql = new StringBuilder();
        // 拼接sql语句
        // insert into user(name,age,sex)values(?,?,?)\
        List<String> fieldsNames = BeanUtil.getFieldsNames(tClass);
        StringBuilder sb1 = new StringBuilder();
        StringBuilder sb2 = new StringBuilder();
        for (String name : fieldsNames){
            sb1.append(name).append(",");
            sb2.append("?,");
        }
        sb1.deleteCharAt(sb1.length()-1);
        sb2.deleteCharAt(sb2.length()-1);

        sql
             .append("insert into ")
                .append(table)
                .append(" ( ")
                //拼接字段名
                .append(sb1.toString())
                .append(" )values( ")
                //拼接占位符
                .append(sb2.toString())
                .append(" )");
       System.out.println(sql.toString());


        return getJdbcSink(driver,url,sql.toString());
    }
    //通过jdbcSink进行封装
    public static <T>SinkFunction<T> getJdbcSink(String driver,String url,String sql){
      return   JdbcSink.sink(
                sql,
                // 占位符赋值
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T t) throws SQLException {
                        //TODO
                        //获取类(三种)：Class.forName("..")  类名.class  对象.getClass
                        Class<?> tClass = t.getClass();
                        Field[] fields = tClass.getDeclaredFields();
                        try {
                            // zengjia yige  jishuqi
                            for (int i = 0, p=1; i < fields.length; i++) {

                                Field field = fields[i];

                                Nosink nosink = field.getAnnotation(Nosink.class);

                                if (nosink == null) {
                                    //private 的属性不允许访问 ,因此添加允许访问私有成员（反射）
                                    field.setAccessible(true);
                                    // 获得 对象的属性名对应的属性值
                                    Object value = field.get(t);
                                    ps.setObject(p++,value);
                                }

                            }
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                           
                        }

                    }

                },
                new JdbcExecutionOptions.Builder()
                        //往外写的间隔时间3s
                        .withBatchIntervalMs(3000)
                        //批次的大小，往外写的条件 1024kb
                        .withBatchSize(1024)
                        //重连次数
                        .withMaxRetries(3)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(driver)
                        .withUrl(url)
                        .build()
        );
    }
}
